package service

import (
	"ao/app/request"
	"context"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	_ "github.com/apache/rocketmq-client-go/v2/admin"
	_ "github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"github.com/gogf/gf/frame/g"
	"github.com/gogf/gf/util/gconv"
	_ "os"
	_ "sync"
)

var nameSrvAddr = []string{"10.99.10.238:9876"}
var topic = "newOne"
var brokerAddr = "10.99.10.238:10911"
var RocketMmq = rocketMmqService{}

type rocketMmqService struct{}

func (s *rocketMmqService)productSendSync(clientReq *request.ClientParams,param interface{})  {
	var err error
	MqProducer, err1 := rocketmq.NewProducer(
		producer.WithGroupName(clientReq.TopicGroup),
		producer.WithNameServer(nameSrvAddr),
		producer.WithRetry(3),
	)
	if err1 != nil {
		panic(fmt.Sprintf("init rocket mq producer err:%v", err1))
		return
	}

	err2 := MqProducer.Start()
	if err2 != nil {
		panic(fmt.Sprintf("producer mq start err:%v", err2))
		return
	}
	defer MqProducer.Shutdown()
	//发送带tag的消息
	body :=gconv.Bytes(param)
	msg := primitive.NewMessage(clientReq.Topic,body)
	msg.WithTag(clientReq.Tag)
	//单向
	err = MqProducer.SendOneWay(context.Background(), msg)
	if err !=nil {
		g.Log().Infof("mq push err :%v",err)
	}

}
